package org.apache.rocketmq.proxy.grpc.v2.common;

import apache.rocketmq.v2.ClientType;
import apache.rocketmq.v2.CustomizedBackoff;
import apache.rocketmq.v2.ExponentialBackoff;
import apache.rocketmq.v2.Settings;
import com.google.protobuf.Duration;
import com.google.protobuf.util.Durations;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.config.ProxyConfig;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.remoting.protocol.subscription.*;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/1/3
 * @方法描述：这个类的对象就是客户端配置管理器，在第四版本代码中，这个类还暂时用不到，大家忽略即可，也不要关心它的具体作用，引入进来只是为了程序不报错，还有一个原因就是如果不引入进来
 * 那么好几个类的构造方法就要简化一下，但是在后面版本很快又要重构成源码那样，为了减少不必要的麻烦，所以我就先引入了
 * 当然，引入进来的这个类肯定是残缺的，等后面引入了生产者和消费者客户端时，这个类就会被重构完整
 */
public class GrpcClientSettingsManager extends ServiceThread implements StartAndShutdown {


    private static final Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
    //客户端配置表，key是客户端唯一Id，value就是对应的客户端配置
    protected static final Map<String, Settings> CLIENT_SETTINGS_MAP = new ConcurrentHashMap<>();

    private final MessagingProcessor messagingProcessor;

    public GrpcClientSettingsManager(MessagingProcessor messagingProcessor) {
        this.messagingProcessor = messagingProcessor;
    }

    //根据客户端Id获取客户端配置
    public Settings getRawClientSettings(String clientId) {
        return CLIENT_SETTINGS_MAP.get(clientId);
    }

    //根据代理上下文对象得到客户端配置的方法
    public Settings getClientSettings(ProxyContext ctx) {
        String clientID = ctx.getClientID();
        Settings settings = getRawClientSettings(clientID);
        if (settings == null){
            return null;
        }
        if (settings.hasPublishing()){
            settings = mergeProducerData(settings);
        }else if (settings.hasSubscription()){
            settings = mergeSubscriptionData(ctx, settings, settings.getSubscription().getGroup().getName());
        }
        return settings;
    }

    //在这个方法中，服务端会把自己的配置信息设置到Settings对象中，这个Settings对象最终会返回给客户端
    //这里大家应该也能意识到了，客户端的设置信息其实是从服务端获得的，现在大家应该知道第九版本遗留问题的答案了吧？
    protected static Settings mergeProducerData(Settings settings) {
        //从服务端的配置管理器中得到配置信息对象
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        //得到Settings对象的构建器
        Settings.Builder builder = settings.toBuilder();
        //向Strrings对象中设置重试策略
        builder.getBackoffPolicyBuilder()
                //设置请求重试最大次数
                .setMaxAttempts(config.getGrpcClientProducerMaxAttempts())
                //接下来设置重试策略的其他参数
                .setExponentialBackoff(ExponentialBackoff.newBuilder()
                        //设置初始退避时间
                        .setInitial(Durations.fromMillis(config.getGrpcClientProducerBackoffInitialMillis()))
                        //设置最大退避时间
                        .setMax(Durations.fromMillis(config.getGrpcClientProducerBackoffMaxMillis()))
                        //设置退避时间指数
                        .setMultiplier(config.getGrpcClientProducerBackoffMultiplier())
                        //构建重试策略
                        .build());

        //设置和发布信息相关的参数
        builder.getPublishingBuilder()
                //设置是否启用消息类型检查的标志
                .setValidateMessageType(config.isEnableTopicMessageTypeCheck())
                //设置可发送消息的最大大小
                .setMaxBodySize(config.getMaxMessageSize());
        //构建Settings对象并返回
        return builder.build();
    }
    //以下方法是消费者客户端的订阅数据的合并方法，现在我们还没有真的实现消费者客户端，所以下面凡是和消费者客户端相关的方法，我就
    //先不添加注释了，大家也不必关注，否则目前要关注的内容就太多了，我们只看文章中涉及到的内容
    protected Settings mergeSubscriptionData(ProxyContext ctx, Settings settings, String consumerGroup) {
        SubscriptionGroupConfig config = this.messagingProcessor.getSubscriptionGroupConfig(ctx, consumerGroup);
        if (config == null) {
            return settings;
        }
        return mergeSubscriptionData(settings, config);
    }

    //该方法和消费则会客户端相关，暂时不添加相关注释
    protected static Settings mergeSubscriptionData(Settings settings, SubscriptionGroupConfig groupConfig) {
        Settings.Builder resultSettingsBuilder = settings.toBuilder();
        ProxyConfig config = ConfigurationManager.getProxyConfig();
        resultSettingsBuilder.getSubscriptionBuilder()
                .setReceiveBatchSize(config.getGrpcClientConsumerLongPollingBatchSize())
                .setLongPollingTimeout(Durations.fromMillis(config.getGrpcClientConsumerMaxLongPollingTimeoutMillis()))
                .setFifo(groupConfig.isConsumeMessageOrderly());


        resultSettingsBuilder.getBackoffPolicyBuilder().setMaxAttempts(groupConfig.getRetryMaxTimes() + 1);
        GroupRetryPolicy groupRetryPolicy = groupConfig.getGroupRetryPolicy();
        if (groupRetryPolicy.getType().equals(GroupRetryPolicyType.EXPONENTIAL)) {
            ExponentialRetryPolicy exponentialRetryPolicy = groupRetryPolicy.getExponentialRetryPolicy();
            if (exponentialRetryPolicy == null) {
                exponentialRetryPolicy = new ExponentialRetryPolicy();
            }
            resultSettingsBuilder.getBackoffPolicyBuilder().setExponentialBackoff(convertToExponentialBackoff(exponentialRetryPolicy));
        } else {
            CustomizedRetryPolicy customizedRetryPolicy = groupRetryPolicy.getCustomizedRetryPolicy();
            if (customizedRetryPolicy == null) {
                customizedRetryPolicy = new CustomizedRetryPolicy();
            }
            resultSettingsBuilder.getBackoffPolicyBuilder().setCustomizedBackoff(convertToCustomizedRetryPolicy(customizedRetryPolicy));
        }
        return resultSettingsBuilder.build();
    }

    //该方法和消费则会客户端相关，暂时不添加相关注释
    protected static ExponentialBackoff convertToExponentialBackoff(ExponentialRetryPolicy retryPolicy) {
        return ExponentialBackoff.newBuilder()
                .setInitial(Durations.fromMillis(retryPolicy.getInitial()))
                .setMax(Durations.fromMillis(retryPolicy.getMax()))
                .setMultiplier(retryPolicy.getMultiplier())
                .build();
    }

    //该方法和消费则会客户端相关，暂时不添加相关注释
    protected static CustomizedBackoff convertToCustomizedRetryPolicy(CustomizedRetryPolicy retryPolicy) {
        List<Duration> durationList = Arrays.stream(retryPolicy.getNext())
                .mapToObj(Durations::fromMillis).collect(Collectors.toList());
        return CustomizedBackoff.newBuilder()
                .addAllNext(durationList)
                .build();
    }

    //更新指定客户端id的客户端配置
    public void updateClientSettings(ProxyContext ctx,String clientId, Settings settings) {
        //如果设置中包含订阅信息，则使用默认的消费者设置构建器合并配置信息
        if (settings.hasSubscription()){
            settings = createDefaultConsumerSettingsBuilder().mergeFrom(settings).build();
        }
        CLIENT_SETTINGS_MAP.put(clientId, settings);
    }
    //创建默认的消费者设置构建器的方法
    protected Settings.Builder createDefaultConsumerSettingsBuilder() {
        return mergeSubscriptionData(Settings.newBuilder().getDefaultInstanceForType(), new SubscriptionGroupConfig()).toBuilder();
    }

    //移除并返回指定客户端Id的客户端配置信息的方法
    public Settings removeAndGetRawClientSettings(String clientId) {
        return CLIENT_SETTINGS_MAP.remove(clientId);
    }


    //根据上下文对象移除指定客户端配置信息的方法
    public Settings removeAndGetClientSettings(ProxyContext ctx) {
        //得到客户端Id
        String clientId = ctx.getClientID();
        //从客户端配置表中移除并得到对应的客户端配置信息
        Settings settings = this.removeAndGetRawClientSettings(clientId);
        //如果设置为null，直接返回null
        if (settings == null) {
            return null;
        }
        //如果设置中包含订阅信息，则合并相关的订阅数据
        if (settings.hasSubscription()) {
            settings = mergeSubscriptionData(ctx, settings, settings.getSubscription().getGroup().getName());
        }
        return settings;
    }
    @Override
    public void run() {
        //该方法的逻辑非常简单，就是在程序运行期间定期判断是否有需要清理的客户端配置信息
        while (!this.isStopped()) {
            //每隔5秒检查一次
            this.waitForRunning(TimeUnit.SECONDS.toMillis(5));
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：检查是否有需要清理的客户端配置信息的方法，如果有那就清理该客户端配置信息，该方法清理的主要是消费者客户端的设置信息，涉及到消费者客户端的一些概念
     * 在第十版本代码中就先不添加注释了
     */
    @Override
    protected void onWaitEnd() {
        Set<String> clientIdSet = CLIENT_SETTINGS_MAP.keySet();
        for (String clientId : clientIdSet) {
            try {
                CLIENT_SETTINGS_MAP.computeIfPresent(clientId, (clientIdKey, settings) -> {
                    if (!settings.getClientType().equals(ClientType.PUSH_CONSUMER) && !settings.getClientType().equals(ClientType.SIMPLE_CONSUMER)) {
                        return settings;
                    }
                    String consumerGroup = settings.getSubscription().getGroup().getName();
                    ConsumerGroupInfo consumerGroupInfo = this.messagingProcessor.getConsumerGroupInfo(
                            ProxyContext.createForInner(this.getClass()),
                            consumerGroup
                    );
                    if (consumerGroupInfo == null || consumerGroupInfo.findChannel(clientId) == null) {
                        log.info("remove unused grpc client settings. group:{}, settings:{}", consumerGroup, settings);
                        return null;
                    }
                    return settings;
                });
            } catch (Throwable t) {
                log.error("check expired grpc client settings failed. clientId:{}", clientId, t);
            }
        }
    }


    @Override
    public String getServiceName() {
        return "GrpcClientSettingsManagerCleaner";
    }

}
